package com.atguigu.flink.chapter06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/22 13:52
 */
public class Flink10_OrderTxDetect {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        // 1.读取数据、转换成 POJO
        SingleOutputStreamOperator<OrderEvent> orderDS = env
                .readTextFile("input/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new OrderEvent(Long.valueOf(datas[0]), datas[1], datas[2], Long.valueOf(datas[3]));
                    }
                });

        SingleOutputStreamOperator<TxEvent> txDS = env
                .readTextFile("input/ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new TxEvent(datas[0], datas[1], Long.valueOf(datas[2]));
                    }
                });

        KeyedStream<OrderEvent, String> orderKS = orderDS.keyBy(order -> order.getTxId());
        KeyedStream<TxEvent, String> txKS = txDS.keyBy(tx -> tx.getTxId());
        // 实时对账 => 对比两条流中的 交易码，如果匹配上，对账成功 => 连接两条流
        ConnectedStreams<OrderEvent, TxEvent> orderTxCS = orderKS.connect(txKS);

//        ConnectedStreams<OrderEvent, TxEvent> orderTxCS = orderDS.connect(txDS);

        orderTxCS
//                .keyBy(order -> order.getTxId(), tx -> tx.getTxId())
                .process(new OrderTxDetectFunction())
                .print();


        env.execute();
    }

    public static class OrderTxDetectFunction extends CoProcessFunction<OrderEvent, TxEvent, String> {
        // 如果 是 tx的数据先处理，那么就 把 它存到一个 map里
        // key为 交易码，value为 数据本身
        Map<String, TxEvent> txEventMap = new HashMap<>();
        // 如果 是 order的数据先处理，那么就 把 它存到一个 map里
        // key为 交易码，value为 数据本身
        Map<String, OrderEvent> orderEventMap = new HashMap<>();

        /**
         * 处理业务系统的数据 OrderEvent
         *
         * @param value
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {
            // 当前是 业务系统的数据， 先查一下，相同 交易码的 交易系统的数据 存不存在
            if (txEventMap.containsKey(value.getTxId())) {
                // 1. 说明 相同交易码 的 交易数据 ，来过 => 匹配成功、对账成功
                out.collect("订单" + value.getOrderId() + "对账成功！");
                // 删除保存的 对应交易码的 交易数据
                txEventMap.remove(value.getTxId());
            } else {
                // 2. 说明 有 相同交易码的 交易数据，没来过 => 把自己（业务数据）存起来
                orderEventMap.put(value.getTxId(), value);
            }
        }

        /**
         * 处理交易系统的数据 TxEvent
         *
         * @param value
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
            // 当前是 交易系统 的数据， 先查一下，相同 交易码的 业务系统的数据 存不存在
            if (orderEventMap.containsKey(value.getTxId())) {
                // 1. 说明 相同交易码 的 业务数据 ，来过 => 匹配成功、对账成功
                out.collect("订单" + orderEventMap.get(value.getTxId()).getOrderId() + "对账成功！");
                // 删除保存的 对应交易码的 业务数据
                orderEventMap.remove(value.getTxId());
            } else {
                // 2. 说明 有 相同交易码的 业务数据，没来过 => 把自己（交易数据）存起来
                txEventMap.put(value.getTxId(), value);
            }
        }
    }
}
/*
    new ProcessFunction
    => 每个 并行实例 都会执行一次， 也就是说 => 每个并行实例，有一个对象， 对应的 属性也只会存一份
    => 所谓的来一条数据处理一条，指的是 方法的调用，  每来一条数据，就会 调用一次 类的对象.方法（）
 */

/*
    注意：
        1、一般 使用 Connect连接两条流，做一些条件匹配的时候， 在多并行度条件下，要根据连接条件keyby
        2、先分别keyby，再connect  跟    先connect，再keyby ，效果一样
 */